#include "access/pg_mirror.h"
#include "postgres.h"
#include "access/xlogrecord.h"
#include "access/heapam_xlog.h"
#include "access/nbtxlog.h"
#include "access/gistxlog.h"
#include "access/spgxlog.h"
#include "access/brin_xlog.h"
#include "assert.h"
#include "common/controldata_utils.h"
#include "miscadmin.h"
#define INSERT_FREESPACE_MIRROR(endptr) \
	(((endptr) % XLOG_BLCKSZ == 0) ? 0 : (XLOG_BLCKSZ - (endptr) % XLOG_BLCKSZ))

static ControlFileData *ControlFile = NULL;
//default 16MB
static int	WalSegSz = 16777216;

//muti block to one record
typedef struct XLogHe3ToPg {
	uint64		CurrBytePos;
	uint64		PrevBytePos;
}XLogHe3ToPg;
static XLogHe3ToPg g_walHe3ToPg;

static void ReConvertMainData(XLogRecord* sRecord, char*sMainData, uint32_t*sLen, char* dMainData, uint32_t* dLen) {
	RmgrId rmid = sRecord->xl_rmid;
    uint8		info = (sRecord->xl_info & ~XLR_INFO_MASK);
	bool hasChange = false;
	switch(rmid) {
	    case RM_HEAP2_ID:
	    {
	    	 if ((info & XLOG_HEAP_OPMASK) == XLOG_HEAP2_VISIBLE) {
			 	xl_heap_visible *xlrec = (xl_heap_visible *)sMainData;
				xl_old_heap_visible xlrecOld;
			 	xlrecOld.cutoff_xid = xlrec->cutoff_xid;
				xlrecOld.flags = xlrec->flags;
				*dLen = sizeof(xl_old_heap_visible);
				memcpy(dMainData,&xlrecOld,*dLen);
				hasChange = true;
	    	 }
			 break;
	    }
		case RM_HEAP_ID:
		{
			if (((info & XLOG_HEAP_OPMASK) == XLOG_HEAP_UPDATE) ||
	                ((info & XLOG_HEAP_OPMASK) == XLOG_HEAP_HOT_UPDATE)) {
                xl_heap_update *xlrec = (xl_heap_update *)sMainData;
            	xl_old_heap_update xlrecOld;
				xlrecOld.old_xmax = xlrec->old_xmax;
                xlrecOld.old_offnum = xlrec->old_offnum;
                xlrecOld.old_infobits_set = xlrec->old_infobits_set;
                xlrecOld.flags = xlrec->flags;
                xlrecOld.new_xmax = xlrec->new_xmax;
                xlrecOld.new_offnum = xlrec->new_offnum;
				*dLen = sizeof(xl_old_heap_update);
				memcpy(dMainData,&xlrecOld,*dLen);
				hasChange = true;
	        }
			break;
		}
		case RM_BTREE_ID:
		{
			if (info == XLOG_BTREE_SPLIT_L || info == XLOG_BTREE_SPLIT_R) {
				xl_btree_split *xlrec = (xl_btree_split *)sMainData;
                xl_old_btree_split xlrecOld;
				xlrecOld.level = xlrec->level;
                xlrecOld.firstrightoff = xlrec->firstrightoff;
                xlrecOld.newitemoff = xlrec->newitemoff;
                xlrecOld.postingoff = xlrec->postingoff;
				*dLen = sizeof(xl_old_btree_split);
				memcpy(dMainData,&xlrecOld,*dLen);
				hasChange = true;
			}
			break;
		}
		case RM_GIST_ID:
		{
			if (info == XLOG_GIST_PAGE_SPLIT) {
				gistxlogPageSplit *xlrec = (gistxlogPageSplit *)sMainData;
                gistoldxlogPageSplit xlrecOld;
				xlrecOld.origrlink = xlrec->origrlink;
				xlrecOld.orignsn = xlrec->orignsn;
				xlrecOld.origleaf = xlrec->origleaf;
				xlrecOld.npage = xlrec->npage;
				xlrecOld.markfollowright = xlrec->markfollowright;
				*dLen = sizeof(gistoldxlogPageSplit);
				memcpy(dMainData,&xlrecOld,*dLen);
				hasChange = true;
			}
			break;
		}
		case RM_SPGIST_ID:
	    {
	        if (info == XLOG_SPGIST_ADD_LEAF) {
				spgxlogAddLeaf *xlrec = (spgxlogAddLeaf *)sMainData;
                spgoldxlogAddLeaf xlrecOld;
				xlrecOld.newPage = xlrec->newPage;		
			    xlrecOld.storesNulls = xlrec->storesNulls;	
			    xlrecOld.offnumLeaf = xlrec->offnumLeaf;	
			    xlrecOld.offnumHeadLeaf = xlrec->offnumHeadLeaf;	
			    xlrecOld.offnumParent = xlrec->offnumParent;	
			    xlrecOld.nodeI = xlrec->nodeI;
				*dLen = sizeof(spgoldxlogAddLeaf);
				memcpy(dMainData,&xlrecOld,*dLen);
				hasChange = true;
	        } else if (info == XLOG_SPGIST_MOVE_LEAFS) {
				spgxlogMoveLeafs *xlrec = (spgxlogMoveLeafs *)sMainData;
                spgoldxlogMoveLeafs xlrecOld;
				xlrecOld.nMoves = xlrec->nMoves;			
			    xlrecOld.newPage = xlrec->newPage;		
			    xlrecOld.replaceDead = xlrec->replaceDead;
			    xlrecOld.storesNulls = xlrec->storesNulls;	
			    xlrecOld.offnumParent = xlrec->offnumParent;
			    xlrecOld.nodeI = xlrec->nodeI;
			    xlrecOld.stateSrc = xlrec->stateSrc;
				*dLen = SizeOfOldSpgxlogMoveLeafs;
				memcpy(dMainData,&xlrecOld,*dLen);
   				memcpy(dMainData+*dLen,xlrec->offsets,*sLen-SizeOfSpgxlogMoveLeafs);	
				*dLen += *sLen-SizeOfSpgxlogMoveLeafs;
				hasChange = true;
	        } else if (info == XLOG_SPGIST_ADD_NODE) {
				spgxlogAddNode *xlrec = (spgxlogAddNode *)sMainData;
                spgoldxlogAddNode xlrecOld;
				xlrecOld.offnum = xlrec->offnum;
			    xlrecOld.offnumNew = xlrec->offnumNew;
			    xlrecOld.newPage = xlrec->newPage;		
			    xlrecOld.parentBlk = xlrec->parentBlk;
			    xlrecOld.offnumParent = xlrec->offnumParent;
			    xlrecOld.nodeI = xlrec->nodeI;
			    xlrecOld.stateSrc = xlrec->stateSrc;
				*dLen = sizeof(spgoldxlogAddNode);
				memcpy(dMainData,&xlrecOld,*dLen);
				hasChange = true;
	        } else if (info == XLOG_SPGIST_PICKSPLIT) {
				spgxlogPickSplit *xlrec = (spgxlogPickSplit *)sMainData;
                spgoldxlogPickSplit xlrecOld;
				xlrecOld.isRootSplit = xlrec->isRootSplit;
				xlrecOld.nDelete = xlrec->nDelete;		
			    xlrecOld.nInsert = xlrec->nInsert;		
			    xlrecOld.initSrc = xlrec->initSrc;		
			    xlrecOld.initDest = xlrec->initDest;		
			    xlrecOld.offnumInner = xlrec->offnumInner;
			    xlrecOld.initInner = xlrec->initInner;		
			    xlrecOld.storesNulls = xlrec->storesNulls;	
			    xlrecOld.innerIsParent = xlrec->innerIsParent;	
			    xlrecOld.offnumParent = xlrec->offnumParent;
			    xlrecOld.nodeI = xlrec->nodeI;
			    xlrecOld.stateSrc = xlrec->stateSrc;
				*dLen = SizeOfOldSpgxlogPickSplit;
				memcpy(dMainData,&xlrecOld,*dLen);
			    memcpy(dMainData+*dLen,xlrec->offsets,*sLen-SizeOfSpgxlogPickSplit);
				*dLen += *sLen-SizeOfSpgxlogPickSplit;
				hasChange = true;
	        }
			break;
		}
		case RM_BRIN_ID:
	    {
	        if (info == XLOG_BRIN_INSERT) {
				xl_brin_insert *xlrec = (xl_brin_insert *)sMainData;
                xl_old_brin_insert xlrecOld;
				xlrecOld.heapBlk = xlrec->heapBlk;
				/* extra information needed to update the revmap */
				xlrecOld.pagesPerRange = xlrec->pagesPerRange;
				xlrecOld.offnum = xlrec->offnum;
				*dLen = sizeof(xl_old_brin_insert);
				memcpy(dMainData,&xlrecOld,*dLen);
				hasChange = true;
	        } else if ( info == XLOG_BRIN_UPDATE) {
				xl_brin_update *xlrec = (xl_brin_update *) sMainData;
                xl_old_brin_update xlrecUpdate;
                xl_brin_insert *xlrecInsert = &xlrec->insert;
                xl_old_brin_insert xlrecOld;
				xlrecOld.heapBlk = xlrecInsert->heapBlk;
				/* extra information needed to update the revmap */
				xlrecOld.pagesPerRange = xlrecInsert->pagesPerRange;
				xlrecOld.offnum = xlrecInsert->offnum;
				/* offset number of old tuple on old page */
				xlrecUpdate.oldOffnum = xlrec->oldOffnum;
				xlrecUpdate.insert = xlrecOld;
				*dLen = sizeof(xl_old_brin_update);
				memcpy(dMainData,&xlrecUpdate,*dLen);
				hasChange = true;
	        }
			break;
		}
		default:
		{
			break;
		}
	}
	if (hasChange == false) {
		*dLen = *sLen;
		memcpy(dMainData,sMainData,*dLen);
	}
}

static int XlogHe3ToPg(XLogRecord*newRecord[],int n, OldXLogRecord*oldRecord) {
	oldRecord->xl_xid = newRecord[0]->xl_xid;
	oldRecord->xl_info = newRecord[0]->xl_info;
	oldRecord->xl_rmid = newRecord[0]->xl_rmid;
	char d_main_data[8192];
	int dPos = 0;
	char* dst = (char*)oldRecord;
	dPos += sizeof(OldXLogRecord);
	uint32_t d_main_data_len = 0;
	uint32	main_data_len = 0;
	uint8_t blkNum = 0;
	bool hasblk = false;
	char*img_ptr[XLR_MAX_BLOCK_ID + 1] = {0};
	char*data_ptr[XLR_MAX_BLOCK_ID + 1] = {0};
	uint16_t bimg_len[XLR_MAX_BLOCK_ID + 1] = {0};
	uint16_t data_len[XLR_MAX_BLOCK_ID + 1] = {0};
	for(int i = 0;i<n;i++) {
		int sPos = 0;
		char* src = (char*)newRecord[i];
		uint32		remaining = newRecord[i]->xl_tot_len - sizeof(XLogRecord);
    	uint32		datatotal = 0;
		sPos += sizeof(XLogRecord);
		while(remaining > datatotal) {
			uint8_t block_id = *(src + sPos);
			if (block_id == XLR_BLOCK_ID_DATA_SHORT) {
				sPos += sizeof(block_id);
				remaining -= sizeof(block_id);
				if (i == n-1) {
					memcpy(dst + dPos,&block_id,sizeof(block_id));
					dPos += sizeof(block_id);
				}
				main_data_len = *((uint8_t*)(src + sPos));
				//main_data_len type XLR_BLOCK_ID_DATA_SHORT
				uint8 d_len;
				if (i == n-1) {
					ReConvertMainData(newRecord[i],src + sPos + sizeof(d_len)+bimg_len[blkNum]+data_len[blkNum],&main_data_len,d_main_data,&d_main_data_len);
					d_len = d_main_data_len;
					memcpy(dst + dPos,&d_len,sizeof(d_len));
					dPos += sizeof(d_len);
				}
				sPos += sizeof(d_len);
				remaining -= sizeof(d_len);
				datatotal += main_data_len;
				break;
			} else if (block_id == XLR_BLOCK_ID_DATA_LONG) {
				sPos += sizeof(block_id);
				remaining -= sizeof(block_id);
				if (i == n-1) {
					memcpy((dst + dPos),&block_id,sizeof(block_id));
					dPos += sizeof(block_id);
				}
				memcpy(&main_data_len,src + sPos,sizeof(uint32));
				if (i == n-1) {
					ReConvertMainData(newRecord[i],src + sPos + sizeof(main_data_len)+bimg_len[blkNum]+data_len[blkNum],&main_data_len,d_main_data,&d_main_data_len);
					if (d_main_data_len > 255) {
						memcpy(dst + dPos,&d_main_data_len,sizeof(d_main_data_len));
						dPos += sizeof(d_main_data_len);
					} else {
						*(dst + dPos - 1) = XLR_BLOCK_ID_DATA_SHORT;
						uint8_t d_len = d_main_data_len;
						memcpy(dst + dPos,&d_len,sizeof(d_len));
						dPos += sizeof(d_len);
					}
				}
				sPos += sizeof(main_data_len);
				remaining -= sizeof(main_data_len);
		        datatotal += main_data_len;
				break;
			} else if (block_id == XLR_BLOCK_ID_ORIGIN) {
				sPos += sizeof(block_id);
				remaining -= sizeof(block_id);
				if (i == n-1) {
					memcpy(dst + dPos,&block_id,sizeof(block_id));
					dPos += sizeof(block_id);
					memcpy(dst + dPos,src+sPos,sizeof(RepOriginId));
					dPos += sizeof(RepOriginId);
				}
				sPos += sizeof(RepOriginId);
				remaining -= sizeof(RepOriginId);
			} else if (block_id == XLR_BLOCK_ID_TOPLEVEL_XID) {
				sPos += sizeof(block_id);
				remaining -= sizeof(block_id);
				if (i == n - 1) {
					memcpy(dst + dPos,&block_id,sizeof(block_id));
					dPos += sizeof(block_id);
					memcpy(dst + dPos,src+sPos,sizeof(TransactionId));
					dPos += sizeof(TransactionId);
				}
				sPos += sizeof(TransactionId);
				remaining -= sizeof(TransactionId);
			} else if (block_id <= XLR_MAX_BLOCK_ID) {
				memcpy(dst + dPos, src + sPos, SizeOfXLogRecordBlockHeader);
				uint8_t		fork_flags = *(src + sPos + sizeof(block_id));
				*(dst + dPos) = blkNum;
				hasblk = true;
				data_len[blkNum] = *((uint16_t*)(src + sPos + sizeof(block_id) + sizeof(fork_flags)));
				datatotal += data_len[blkNum];
				sPos += SizeOfXLogRecordBlockHeader;
				dPos += SizeOfXLogRecordBlockHeader;
				remaining -= SizeOfXLogRecordBlockHeader;
				if ((fork_flags & BKPBLOCK_HAS_IMAGE) != 0) {
					bimg_len[blkNum] = *((uint16_t*)(src + sPos));
					datatotal += bimg_len[blkNum];
					uint16_t hole_offset = *((uint16_t*)(src + sPos + sizeof(bimg_len)));
					uint8_t bimg_info = *((uint16_t*)(src + sPos + sizeof(bimg_len) + sizeof(hole_offset)));
					memcpy(dst + dPos, src + sPos, SizeOfXLogRecordBlockImageHeader);
					sPos += SizeOfXLogRecordBlockImageHeader;
					dPos += SizeOfXLogRecordBlockImageHeader;
					remaining -= SizeOfXLogRecordBlockImageHeader;
					if ((bimg_info & BKPIMAGE_IS_COMPRESSED) != 0) {
						if ((bimg_info & BKPIMAGE_HAS_HOLE) != 0) {
							memcpy(dst + dPos, src + sPos, SizeOfXLogRecordBlockCompressHeader);
							sPos += SizeOfXLogRecordBlockCompressHeader;
							dPos += SizeOfXLogRecordBlockCompressHeader;
							remaining -= SizeOfXLogRecordBlockCompressHeader;
						}
					}
				}
				if (!(fork_flags & BKPBLOCK_SAME_REL)) {
					 memcpy(dst + dPos, src + sPos, sizeof(RelFileNode));
					 sPos += sizeof(RelFileNode);
					 dPos += sizeof(RelFileNode);
					 remaining -= sizeof(RelFileNode);
				} 
				memcpy(dst + dPos, src + sPos, sizeof(BlockNumber));
				sPos += sizeof(BlockNumber);
				dPos += sizeof(BlockNumber);
				remaining -= sizeof(BlockNumber);
			} else {
				printf("invalid block_id %u",block_id);
			}
		}
		assert(remaining == datatotal);
		if (bimg_len[blkNum] != 0 ) {
			img_ptr[blkNum] = src + sPos;
			sPos += bimg_len[blkNum];
		}
		if (data_len[blkNum] != 0) {
			data_ptr[blkNum] = src + sPos;
			sPos += data_len[blkNum];
		}
		if (hasblk == true) {
			blkNum++;
		}
		
		sPos += main_data_len;
		assert(sPos == newRecord[i]->xl_tot_len);
	}
	int idx = 0;
	while(idx < blkNum) {
		if (bimg_len[idx] != 0) {
			memcpy(dst + dPos, img_ptr[idx], bimg_len[idx]);
			dPos += bimg_len[idx];
		}
		if (data_len[idx] != 0){
			memcpy(dst + dPos, data_ptr[idx], data_len[idx]);
			dPos += data_len[idx];
		}
		idx++;
	}
	memcpy(dst + dPos, d_main_data, d_main_data_len);
	dPos += d_main_data_len;
	oldRecord->xl_tot_len = dPos;
	return dPos;
}

static int OldUsableBytesInSegment =
		(DEFAULT_XLOG_SEG_SIZE / XLOG_BLCKSZ * (XLOG_BLCKSZ - SizeOfXLogShortPHD)) -
		(SizeOfXLogLongPHD - SizeOfXLogShortPHD);


static XLogRecPtr
OldXLogBytePosToRecPtr(uint64 bytepos)
{
	uint64		fullsegs;
	uint64		fullpages;
	uint64		bytesleft;
	uint32		seg_offset;
	XLogRecPtr	result;

	fullsegs = bytepos / OldUsableBytesInSegment;
	bytesleft = bytepos % OldUsableBytesInSegment;

	if (bytesleft < XLOG_BLCKSZ - SizeOfXLogLongPHD)
	{
		/* fits on first page of segment */
		seg_offset = bytesleft + SizeOfXLogLongPHD;
	}
	else
	{
		/* account for the first page on segment with long header */
		seg_offset = XLOG_BLCKSZ;
		bytesleft -= XLOG_BLCKSZ - SizeOfXLogLongPHD;

		fullpages = bytesleft / (XLOG_BLCKSZ - SizeOfXLogShortPHD);
		bytesleft = bytesleft % (XLOG_BLCKSZ - SizeOfXLogShortPHD);

		seg_offset += fullpages * XLOG_BLCKSZ + bytesleft + SizeOfXLogShortPHD;
	}

	XLogSegNoOffsetToRecPtr(fullsegs, seg_offset, DEFAULT_XLOG_SEG_SIZE, result);

	return result;
}

static XLogRecPtr
OldXLogBytePosToEndRecPtr(uint64 bytepos)
{
	uint64		fullsegs;
	uint64		fullpages;
	uint64		bytesleft;
	uint32		seg_offset;
	XLogRecPtr	result;

	fullsegs = bytepos / OldUsableBytesInSegment;
	bytesleft = bytepos % OldUsableBytesInSegment;

	if (bytesleft < XLOG_BLCKSZ - SizeOfXLogLongPHD)
	{
		/* fits on first page of segment */
		if (bytesleft == 0)
			seg_offset = 0;
		else
			seg_offset = bytesleft + SizeOfXLogLongPHD;
	}
	else
	{
		/* account for the first page on segment with long header */
		seg_offset = XLOG_BLCKSZ;
		bytesleft -= XLOG_BLCKSZ - SizeOfXLogLongPHD;

		fullpages = bytesleft / (XLOG_BLCKSZ - SizeOfXLogShortPHD);
		bytesleft = bytesleft % (XLOG_BLCKSZ - SizeOfXLogShortPHD);

		if (bytesleft == 0)
			seg_offset += fullpages * XLOG_BLCKSZ + bytesleft;
		else
			seg_offset += fullpages * XLOG_BLCKSZ + bytesleft + SizeOfXLogShortPHD;
	}

	XLogSegNoOffsetToRecPtr(fullsegs, seg_offset, DEFAULT_XLOG_SEG_SIZE, result);

	return result;
}

static uint64
OldXLogRecPtrToBytePos(XLogRecPtr ptr)
{
	uint64		fullsegs;
	uint32		fullpages;
	uint32		offset;
	uint64		result;

	XLByteToSeg(ptr, fullsegs, DEFAULT_XLOG_SEG_SIZE);

	fullpages = (XLogSegmentOffset(ptr, DEFAULT_XLOG_SEG_SIZE)) / XLOG_BLCKSZ;
	offset = ptr % XLOG_BLCKSZ;

	if (fullpages == 0)
	{
		result = fullsegs * OldUsableBytesInSegment;
		if (offset > 0)
		{
			Assert(offset >= SizeOfXLogLongPHD);
			result += offset - SizeOfXLogLongPHD;
		}
	}
	else
	{
		result = fullsegs * OldUsableBytesInSegment +
			(XLOG_BLCKSZ - SizeOfXLogLongPHD) + /* account for first page */
			(fullpages - 1) * (XLOG_BLCKSZ - SizeOfXLogShortPHD);	/* full pages */
		if (offset > 0)
		{
			Assert(offset >= SizeOfXLogShortPHD);
			result += offset - SizeOfXLogShortPHD;
		}
	}

	return result;
}

static bool
ReserveXLogWalSwitch(XLogRecPtr *StartPos, XLogRecPtr *EndPos, XLogRecPtr *PrevPtr)
{
	XLogHe3ToPg *Insert = &g_walHe3ToPg;
	uint64		startbytepos;
	uint64		endbytepos;
	uint64		prevbytepos;
	uint32		size = MAXALIGN(SizeOfOldXLogRecord);
	XLogRecPtr	ptr;
	uint32		segleft;

	startbytepos = Insert->CurrBytePos;

	ptr = OldXLogBytePosToEndRecPtr(startbytepos);
	if (XLogSegmentOffset(ptr, DEFAULT_XLOG_SEG_SIZE) == 0)
	{
		*EndPos = *StartPos = ptr;
		return false;
	}

	endbytepos = startbytepos + size;
	prevbytepos = Insert->PrevBytePos;

	*StartPos = OldXLogBytePosToRecPtr(startbytepos);
	*EndPos = OldXLogBytePosToEndRecPtr(endbytepos);

	segleft = DEFAULT_XLOG_SEG_SIZE - XLogSegmentOffset(*EndPos, DEFAULT_XLOG_SEG_SIZE);
	if (segleft != DEFAULT_XLOG_SEG_SIZE)
	{
		/* consume the rest of the segment */
		*EndPos += segleft;
		endbytepos = OldXLogRecPtrToBytePos(*EndPos);
	}
	Insert->CurrBytePos = endbytepos;
	Insert->PrevBytePos = startbytepos;

	*PrevPtr = OldXLogBytePosToRecPtr(prevbytepos);

	Assert(XLogSegmentOffset(*EndPos, DEFAULT_XLOG_SEG_SIZE) == 0);
	Assert(OldXLogRecPtrToBytePos(*EndPos) == endbytepos);
	Assert(OldXLogRecPtrToBytePos(*StartPos) == startbytepos);
	Assert(OldXLogRecPtrToBytePos(*PrevPtr) == prevbytepos);

	return true;
}

static void
ReserveXLogWalInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos,
						  XLogRecPtr *PrevPtr)
{
	XLogHe3ToPg *Insert = &g_walHe3ToPg;
	uint64		startbytepos;
	uint64		endbytepos;
	uint64		prevbytepos;

	size = MAXALIGN(size);

	/* All (non xlog-switch) records should contain data. */
	Assert(size > SizeOfOldXLogRecord);

	/*
	 * The duration the spinlock needs to be held is minimized by minimizing
	 * the calculations that have to be done while holding the lock. The
	 * current tip of reserved WAL is kept in CurrBytePos, as a byte position
	 * that only counts "usable" bytes in WAL, that is, it excludes all WAL
	 * page headers. The mapping between "usable" byte positions and physical
	 * positions (XLogRecPtrs) can be done outside the locked region, and
	 * because the usable byte position doesn't include any headers, reserving
	 * X bytes from WAL is almost as simple as "CurrBytePos += X".
	 */

	startbytepos = Insert->CurrBytePos;
	endbytepos = startbytepos + size;
	prevbytepos = Insert->PrevBytePos;
	Insert->CurrBytePos = endbytepos;
	Insert->PrevBytePos = startbytepos;

	*StartPos = OldXLogBytePosToRecPtr(startbytepos);
	*EndPos = OldXLogBytePosToEndRecPtr(endbytepos);
	*PrevPtr = OldXLogBytePosToRecPtr(prevbytepos);

	/*
	 * Check that the conversions between "usable byte positions" and
	 * XLogRecPtrs work consistently in both directions.
	 */
	Assert(OldXLogRecPtrToBytePos(*StartPos) == startbytepos);
	Assert(OldXLogRecPtrToBytePos(*EndPos) == endbytepos);
	Assert(OldXLogRecPtrToBytePos(*PrevPtr) == prevbytepos);
}

static void CopyXLogRecordToPgWAL(int write_len,OldXLogRecord* rechdr,XLogRecPtr StartPos, XLogRecPtr EndPos,
char*dBuf,int* dLen) {
	
	char	   *currpos;
	int 		freespace;
	int 		written;
	XLogRecPtr	CurrPos;
	XLogPageHeader pagehdr;
	CurrPos = StartPos;
	XLogPageHeader page;
	XLogLongPageHeader longpage;
	currpos = dBuf;
	if (CurrPos % XLOG_BLCKSZ == SizeOfXLogShortPHD &&
		XLogSegmentOffset(CurrPos, DEFAULT_XLOG_SEG_SIZE) > XLOG_BLCKSZ) {
		page = (XLogPageHeader)currpos;
		page->xlp_magic = XLOG_PAGE_MAGIC;
		page->xlp_info = 0;
		page->xlp_tli = ControlFile->checkPointCopy.ThisTimeLineID;
		page->xlp_pageaddr = CurrPos - (CurrPos % XLOG_BLCKSZ);
		currpos += SizeOfXLogShortPHD;
	}
	else if (CurrPos % XLOG_BLCKSZ == SizeOfXLogLongPHD &&
		XLogSegmentOffset(CurrPos, DEFAULT_XLOG_SEG_SIZE) < XLOG_BLCKSZ) {
		page = (XLogPageHeader)currpos;
		page->xlp_magic = XLOG_PAGE_MAGIC;
		page->xlp_info = XLP_LONG_HEADER;
		page->xlp_tli = ControlFile->checkPointCopy.ThisTimeLineID;
		page->xlp_pageaddr = CurrPos - (CurrPos % XLOG_BLCKSZ);
		longpage = (XLogLongPageHeader) page;
		longpage->xlp_sysid = ControlFile->system_identifier;
		longpage->xlp_seg_size = WalSegSz;
		longpage->xlp_xlog_blcksz = XLOG_BLCKSZ;
		currpos += SizeOfXLogLongPHD;
	}	
	
	freespace = INSERT_FREESPACE_MIRROR(CurrPos);
	Assert(freespace >= sizeof(uint32));
	/* Copy record data */
	written = 0;
	if (rechdr != NULL) {
		char	   *rdata_data = rechdr;
		int			rdata_len = rechdr->xl_tot_len;
		while (rdata_len > freespace)
		{
			Assert(CurrPos % XLOG_BLCKSZ >= SizeOfXLogShortPHD || freespace == 0);
			memcpy(currpos, rdata_data, freespace);
			rdata_data += freespace;
			rdata_len -= freespace;
			written += freespace;
			CurrPos += freespace;
			currpos += freespace;
			
			pagehdr = (XLogPageHeader) currpos;
			pagehdr->xlp_info = 0;
			pagehdr->xlp_tli = ControlFile->checkPointCopy.ThisTimeLineID;
			pagehdr->xlp_magic = XLOG_PAGE_MAGIC;
			pagehdr->xlp_pageaddr = CurrPos - (CurrPos % XLOG_BLCKSZ);
			pagehdr->xlp_rem_len = write_len - written;
			pagehdr->xlp_info |= XLP_FIRST_IS_CONTRECORD;
			if (XLogSegmentOffset(CurrPos, DEFAULT_XLOG_SEG_SIZE) == 0) {
				CurrPos += SizeOfXLogLongPHD;
				currpos += SizeOfXLogLongPHD;
				pagehdr->xlp_info |= XLP_LONG_HEADER;
				longpage = (XLogLongPageHeader) pagehdr;
				longpage->xlp_sysid = ControlFile->system_identifier;
				longpage->xlp_seg_size = WalSegSz;
				longpage->xlp_xlog_blcksz = XLOG_BLCKSZ;
			} else {
				CurrPos += SizeOfXLogShortPHD;
				currpos += SizeOfXLogShortPHD;
			}
			freespace = INSERT_FREESPACE_MIRROR(CurrPos);
		}
		Assert(CurrPos % XLOG_BLCKSZ >= SizeOfXLogShortPHD || rdata_len == 0);
		memcpy(currpos, rdata_data, rdata_len);
		currpos += rdata_len;
		CurrPos += rdata_len;
		freespace -= rdata_len;
		written += rdata_len;
	}
	Assert(written == write_len);
	int extra_space = MAXALIGN64(CurrPos) - CurrPos;
	CurrPos = MAXALIGN64(CurrPos);
	if (CurrPos != EndPos)
		printf("ERROR space reserved for WAL record does not match what was written");
	currpos += extra_space;
	*dLen = (int)(currpos - dBuf);
}

void readControlFile(char*pathstr) {
	bool		crc_ok;
	
	ControlFile = get_controlfile(pathstr,&crc_ok);
	if (!crc_ok)
		printf(_("WARNING: Calculated CRC checksum does not match value stored in file.\n"
				 "Either the file is corrupt, or it has a different layout than this program\n"
				 "is expecting.  The results below are untrustworthy.\n\n"));

	/* set wal segment size */
	WalSegSz = ControlFile->xlog_seg_size;
}

void setControlFile(ControlFileData *cfile) {
	ControlFile = cfile;
}


int ArrayXlogHe3ToPg(char*sBuf,int sLen, char*dBuf,int* dLen,uint64 *startLsn,uint64 *endLsn) {
	XLogRecord*one = (XLogRecord*)sBuf;
	//32kB
	static char tBuf[32768];
	int tLen = 0;
	int MtrLen = 0;
	int iLen = 0;
	int oLen = 0;
	*dLen = 0;
	for(;iLen<sLen;) {
		int n = 0;
		XLogRecord*newRecord[XLR_MAX_BLOCK_ID + 1];
		while(one->mtr == false) {
			newRecord[n++] = one;
			iLen += one->xl_tot_len;
			one = (((char*)one) + one->xl_tot_len);
			if (iLen > sLen) {
				break;
		    }
		} 
		newRecord[n++] = one;
		iLen += one->xl_tot_len;
		one = (((char*)one) + one->xl_tot_len);
		if (iLen > sLen) {
			break;
	    }
		XlogHe3ToPg(newRecord,n,tBuf+tLen);
		uint64 StartPos,EndPos;
		XLogRecPtr reduceV = 0;
		if (g_walHe3ToPg.PrevBytePos == 0) {
			uint64 xl_prev = newRecord[0]->xl_end - newRecord[0]->xl_tot_len;
			g_walHe3ToPg.PrevBytePos = g_walHe3ToPg.CurrBytePos = xl_prev;
			bool Insert = ReserveXLogWalSwitch(&StartPos,&EndPos,&xl_prev);
			g_walHe3ToPg.PrevBytePos = g_walHe3ToPg.CurrBytePos;
			reduceV = 1;
		}
		OldXLogRecord* rechdr = (OldXLogRecord*)(tBuf + tLen);
		ReserveXLogWalInsertLocation(rechdr->xl_tot_len,&StartPos,&EndPos,&rechdr->xl_prev);
		//for pg check
		if (rechdr->xl_rmid == RM_XLOG_ID &&
			(rechdr->xl_info == XLOG_CHECKPOINT_SHUTDOWN || rechdr->xl_info == XLOG_CHECKPOINT_ONLINE)) {
			CheckPoint*cp = (CheckPoint*)(((char*)rechdr)+SizeOfOldXLogRecord + SizeOfXLogRecordDataHeaderShort);
			cp->redo = StartPos;
			rechdr->xl_prev = rechdr->xl_prev-reduceV;
		}
		pg_crc32c	rdata_crc;
		INIT_CRC32C(rdata_crc);
		COMP_CRC32C(rdata_crc, ((char*)rechdr) + SizeOfOldXLogRecord, rechdr->xl_tot_len - SizeOfOldXLogRecord);
		COMP_CRC32C(rdata_crc, rechdr, offsetof(OldXLogRecord, xl_crc));
		FIN_CRC32C(rdata_crc);
		rechdr->xl_crc = rdata_crc;
		CopyXLogRecordToPgWAL(rechdr->xl_tot_len,rechdr,StartPos,EndPos,dBuf+*dLen,&oLen);
		if (*startLsn == 0) {
			*startLsn = StartPos;
		}
		*endLsn = EndPos;
		*dLen += oLen;
		tLen += rechdr->xl_tot_len;
		MtrLen = iLen;
	} 
	return MtrLen;
}



